package com.hsylient.kafka.service;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class WaitTaskFinishService {

    @Resource
    private ExecutorService executorService;

    // 很多时候在一次请求中，我们想将一部分任务并行执行，然后等到并行任务都执行完成了，将所有并行任务的结果返回。

    /**
     * 使用Stream的并行流处理列表中的任务，每个任务都会阻塞不同的随机时间
     * 如果并行流成功执行，那么最终这个请求的耗时应该跟最大任务耗时差不多。
     * 结果：满足预期，接口耗时10秒，任务最大耗时也将近10秒
     * 接口调用时间：11:03:36.644
     * 接口返回时间：11:03:46.320
     * 打印的最大耗时：9669
     */
    public void methodByParallelStream() {
        log.info("[ParallelStream测试等待并行任务执行结束] 任务开始.....");
        List<Integer> integers = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        List<Integer> resultList = integers.parallelStream().map(item -> {
            long consumingTime = RandomUtil.randomLong(10000L);
            log.info("[ParallelStream测试等待并行任务执行结束] 当前线程预计要耗时 {}", consumingTime);
            ThreadUtil.sleep(consumingTime);
            return item + 1;
        }).collect(Collectors.toList());
        log.info("[ParallelStream测试等待并行任务执行结束] 任务结束.....");
        log.info("[ParallelStream测试等待并行任务执行结束] 任务执行结束后的集合 {}", resultList);
    }


    /**
     * 结果：满足预期，接口耗时8秒，任务最大耗时也将近7.4秒
     * 接口调用时间：13:47:51.003
     * 接口返回时间：13:47:59.385
     * 打印的最大耗时：7406
     */
    public void methodCompletableFuture() {
        log.info("[CompletableFuture测试等待并行任务执行结束] 任务开始.....");
        List<Integer> integers = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        List<CompletableFuture<Integer>> completableFutureList = integers.stream().map(item -> CompletableFuture.supplyAsync(() ->{
            long consumingTime = RandomUtil.randomLong(10000L);
            log.info("[CompletableFuture测试等待并行任务执行结束] 当前线程预计要耗时 {}", consumingTime);
            ThreadUtil.sleep(consumingTime);
            return item + 1;
        }, executorService)).collect(Collectors.toList());


        CompletableFuture.allOf(completableFutureList.toArray(new CompletableFuture[0])).join();

        List<Integer> result = completableFutureList.stream().map(CompletableFuture::join).collect(Collectors.toList());

        log.info("[CompletableFuture测试等待并行任务执行结束] 所有任务已完成: {}", result);

    }


    /**
     * 结果：满足预期，接口耗时10秒，任务最大耗时也将近10秒
     * 接口调用时间：14:03:02.604
     * 接口返回时间：14:03:12.026
     * 打印的最大耗时：9418
     */
    public void methodExecutorCompletion() {
        log.info("[ExecutorCompletion测试等待并行任务执行结束] 任务开始.....");
        List<Integer> integers = List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        ExecutorCompletionService<Integer> service = new ExecutorCompletionService<>(executorService);
        integers.forEach(item -> service.submit(() -> {
            long consumingTime = RandomUtil.randomLong(10000L);
            log.info("[ExecutorCompletion测试等待并行任务执行结束] 当前线程预计要耗时 {}", consumingTime);
            ThreadUtil.sleep(consumingTime);
            return item + 1;
        }));

        List<Integer> result = new ArrayList<>();
        integers.forEach(item -> {
            try {
                Future<Integer> take = service.take();
                result.add(take.get());
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
        });

        log.info("[ExecutorCompletion测试等待并行任务执行结束] 所有任务已完成: {}", result);
    }
}
